package com.atguigu.userprofile.app

import com.atguigu.userprofile.common.bean.{TagInfo, TaskInfo, TaskTagRule}
import com.atguigu.userprofile.common.constant.ConstCode
import com.atguigu.userprofile.common.dao.{TagInfoDAO, TaskInfoDAO, TaskTagRuleDAO}
import com.atguigu.userprofile.common.util.PropertiesUtil
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

import java.util.Properties


object TaskSQL {

  /*
 1   接收 spark-submit中传递的业务参数  也就是 task_id
 2   根据task_id 查询mysql ，获得对应的task_info \tag_info\ task_tag_rule 信息
 3   获得标签名称 、标签编码、 定义的sql 、 获得计算值与标签值得映射关系
 4   生成标签前，先要检查是否有该标签的表，如果没有，要为该标签建立数据表。
 5   要根据sql和映射，动态拼接一个insert select SQL
 6      执行该sql

 */
  private val properties: Properties =  PropertiesUtil.load("config.properties")


  val HDFS_PATH = properties.getProperty("hdfs-store.path")
  val UP_DBNAME = properties.getProperty("user-profile.dbname")
  val DW_DBNAME = properties.getProperty("data-warehouse.dbname")

  def main(args: Array[String]): Unit = {
    //0 执行环境
    val sparkConf: SparkConf = new SparkConf().setAppName("task_sql_app").setMaster("local[*]")
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()


    //        1   接收 spark-submit中传递的业务参数  也就是 task_id   //约定第一个参数放task_id  第二参数 放busi_date
    //   spark-submit –xx –xx –xx   --class  xxx     xxx.jar   101 2020-06-14 ... ...
     val taskId: String = args(0)
     val busiDate: String = args(1)

    //        2   根据task_id 查询mysql ，获得对应的task_info \tag_info\ task_tag_rule 信息

    val tagInfo: TagInfo = TagInfoDAO.getTagInfoByTaskId(taskId)
    // 2.2 补充 taskInfo 和 taskTagRuleList的查询方法
    val taskInfo:TaskInfo=TaskInfoDAO.getTaskInfoById(taskId)
    val taskTagRuleList:List[TaskTagRule]=TaskTagRuleDAO.getTaskTagRuleListByTaskId(taskId)

    println(tagInfo)
    println(taskInfo)
    println(taskTagRuleList)

    //        3   获得标签名称 、标签编码、 定义的sql 、 获得计算值与标签值得映射关系
    val tagCode=tagInfo.tagCode
    val tagName=tagInfo.tagName
    var tagSQL=taskInfo.taskSql

    //############# 替换 SQL 中的$dt 为业务日期
    tagSQL=tagSQL.replace("$dt",busiDate)

    //###################


    //        4   生成标签前，先要检查是否有该标签的表，如果没有，要为该标签建立数据表。
    // 每张表 负责一个标签   两个字段 谁 ，值是什么 比如 性别标签表  uid,tag_value  分区字段  busiDate
    //   create table  tableName （用tagCode)
    //   (uid string, tag_value  tagValueType?)
    //   partitioned by (dt string)
    //   ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'
    //    location  hdfs标签主目录/画像库名/表名

     val tagValueType = tagInfo.tagValueType  match {
       case ConstCode.TAG_VALUE_TYPE_LONG =>"BIGINT"
       case ConstCode.TAG_VALUE_TYPE_DECIMAL =>"DECIMAL(16,2)"
       case ConstCode.TAG_VALUE_TYPE_STRING =>"STRING"
       case ConstCode.TAG_VALUE_TYPE_DATE =>"STRING"
     }
    val tableName= tagCode.toLowerCase()

    val createTableSQL=
      s"""
         | create  table if not exists ${UP_DBNAME}.${tableName}
         |       (uid string, tag_value  $tagValueType)
         |        partitioned by (dt string)
         |      ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'
         |         location  '${HDFS_PATH}/${UP_DBNAME}/${tableName}'
         |""".stripMargin

    println(createTableSQL)
    sparkSession.sql(createTableSQL)

    //        5   要根据sql和映射，动态拼接一个insert select SQL
    //select uid ,  case  query_value  when  'F' then '女' when 'M’ then '男'....
    //(
     //select id as   uid , if( gender <>'', gender,'U') as  query_value
    // from dim_user_zip  where dt='9999-12-31'
   //)
       // 5.1 把 tasktagRule 转换为 case when 子句

    //如果有四级标签，通过case when 完成查询值到标签值的 映射转换
    //如果无四级标签，查询值即为标签值
    var queryValueSql=""
    if(taskTagRuleList.size>0){
      val whenThenSQLList: List[String] = taskTagRuleList.map(taskTagRule => s"when '${taskTagRule.queryValue}' then '${taskTagRule.subTagValue}'")
      val whenThenSQL: String = whenThenSQLList.mkString(" ")
      val caseSQL=s" case query_value $whenThenSQL end as query_value"
      queryValueSql=caseSQL
    }else{
      queryValueSql =" query_value "
    }


     //5.2 拼接 select 语句
    val selectSQL=
      s"""
         |select uid, $queryValueSql
         |from ( $tagSQL
         |)
         |""".stripMargin

    //5.3 拼接 insert select语句  //默认要在数仓库执行 ，保证画像表必须带库名
    val insertSQL=s" insert overwrite table $UP_DBNAME.$tableName  partition (dt='$busiDate') $selectSQL"


    println(insertSQL)
    //        6      执行该sql  driver? ex?
    sparkSession.sql(s"use $DW_DBNAME")
    sparkSession.sql(insertSQL)


  }

}
